package com.dc.adaptor.transport.queue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import com.dc.adaptor.passive.Message;
import com.dc.adaptor.passive.QueueFactory;
import com.dc.adaptor.transport.ReceiveCallback;
import com.dc.adaptor.transport.TransportMeta;
import com.dc.adaptor.transport.TransportReceiver;

public class QueueReceiver implements TransportReceiver{
	private QueueMeta meta;
	@Override
	public void dispose() {	
	}
	public QueueReceiver(TransportMeta meta){
		this.meta = (QueueMeta)meta;
	}

	@Override
	public boolean receive(ReceiveCallback receiveCallback) {
		Message message = null;
		try {
			String queueId = meta.getQueue();
			BlockingQueue<Message> queue = QueueFactory.getQueue(queueId);
			message = queue.poll(2,TimeUnit.SECONDS);
			if(message == null){
				return false;
			}
			String text = message.getMessage();
			if (receiveCallback != null) {
				if (message.getGlobalBusinessid() != null)
					receiveCallback.beforeCommit(text, message.getGlobalBusinessid());
				else receiveCallback.beforeCommit(text);
			}
			if (message.getQueue() != null) {
				message.getQueue().add("0K");
			}
			if (receiveCallback != null) {
				if (message.getGlobalBusinessid() != null)
					receiveCallback.afterCommit(text, message.getGlobalBusinessid());
				else receiveCallback.afterCommit(text);
			}
		}catch (Exception e) {
			throw new RuntimeException(e);
		}
		return true;
	}

}
